

namespace Stee.CAP8.MessageBus.Kafka.Subscriber
{
    using System;
    using System.Collections.Generic;
    using System.Diagnostics;
    using System.Threading;
    using Confluent.Kafka;
    using NLog;
    using Stee.CAP8.MessageBus.Interface.Delegates;
    using Stee.CAP8.MessageBus.Interface.Models;
    using Stee.CAP8.MessageBus.Interface.Utility;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.ResourceLib;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Poll consumption implementation
    /// </summary>
    /// <typeparam name="TMessage">Message data type</typeparam>
    public class PollingImpl<TMessage> : IDisposable
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        /// <summary>
        /// Timer interval for re-subscription
        /// </summary>
        private const int TimerIntervalSubscribe = 5000;

        #endregion Constant & Static Variables


        #region Fields

        /// <summary>
        /// Handler for processing incoming messages
        /// </summary>
        private readonly HandleSubscribeTopic<TMessage> subscribeMessageHandler;

        /// <summary>
        /// Subscriber error handler
        /// </summary>
        private readonly HandleBrokerError subscribeErrorHandler;

        /// <summary>
        /// Consumer configuration
        /// </summary>
        private readonly ConsumerConfig consumerConfig;

        /// <summary>
        /// Kafka consumer builder
        /// </summary>
        private ConsumerBuilder<string, byte[]> consumerBuilder;

        /// <summary>
        /// Kafka consumer
        /// </summary>
        private IConsumer<string, byte[]> consumer;

        /// <summary>
        /// Cancel token for 
        /// </summary>
        private CancellationTokenSource cancellationToken;

        /// <summary>
        /// Flag indicating polling task is cancelled
        /// </summary>
        private bool isCancelled = false;

        /// <summary>
        /// Flag indicating polling restart
        /// </summary>
        private bool isRestartPolling = false;

        /// <summary>
        /// Timer for re-subscribing
        /// </summary>
        private readonly System.Timers.Timer resubscribeTimer;

        /// <summary>
        /// Mutex lock
        /// </summary>
        private readonly object mutex = new object();

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the message topic
        /// Mandatory configuration.
        /// </summary>
        public TopicName TopicName { get; }

        /// <summary>
        /// Gets the client group id string.
        /// All clients sharing the same group.id belong to the same group.
        /// default: '' importance: high
        /// Mandatory configuration.
        /// </summary>
        public string GroupID
        {
            get { return (this.consumerConfig == null) ? string.Empty : this.consumerConfig.GroupId; }
        }

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID { get; private set; } = string.Empty;

        /// <summary>
        /// Gets the flag indicating if consumer is currently consuming messages
        /// </summary>
        public bool IsConnected { get; private set; }

        /// <summary>
        /// Gets the flag indicating if publisher is disposed
        /// </summary>
        public bool IsDisposed { get; private set; } = false;

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the PollingTask class
        /// </summary>
        /// <param name="topicName">Subscribe topic name</param>
        /// <param name="subscribeTopicHandler">Subscribe message handler, message data type</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        public PollingImpl(TopicName topicName, HandleSubscribeTopic<TMessage> subscribeTopicHandler, HandleBrokerError brokerErrorHandler)
        {
            ValidateHelper.ValidateTopic(topicName);

            if (subscribeTopicHandler == null)
            {
                throw new ArgumentNullException($"{Kafka_Consumer.KAFKA_WARN_NULL_MESSAGEHANDLER}: {nameof(subscribeTopicHandler)}");
            }
            else if (brokerErrorHandler == null)
            {
                throw new ArgumentNullException($"{Kafka_Consumer.KAFKA_WARN_NULL_ERROR_HANDLER}: {nameof(brokerErrorHandler)}");
            }

            this.TopicName = topicName;
            this.subscribeMessageHandler = subscribeTopicHandler;
            this.subscribeErrorHandler = brokerErrorHandler;

            Dictionary<string, string> configs = ConfigurationHelper.ReadJSONConfiguration(ConfigurationHelper.MessageBusSubscribeConfig);
            SubscriberConfiguration configuration = new SubscriberConfiguration(configs);

            this.consumerConfig = new ConsumerConfig
            {
                BootstrapServers = configuration.BootstrapServers,
                GroupId = configuration.GroupID,
                ClientId = configuration.ClientID,
                AutoOffsetReset = configuration.AutoOffsetReset,
                EnableAutoCommit = configuration.EnableAutoCommit,
                EnableAutoOffsetStore = configuration.EnableAutoOffsetStore,
                MaxPollIntervalMs = configuration.MaxPollIntervalMs,
                SessionTimeoutMs = configuration.SessionTimeoutMs,
                AutoCommitIntervalMs = configuration.AutoCommitIntervalMs,
                CheckCrcs = configuration.CheckCrcs,
                PartitionAssignmentStrategy = configuration.PartitionAssignmentStrategy,
            };

            //// Create re-subscribe timer
            this.resubscribeTimer = new System.Timers.Timer(TimerIntervalSubscribe);
            this.resubscribeTimer.Elapsed += this.ResubscribeTimerOnElapsed;

            //// Initialize consumer builder
            this.InitializeConsumer();
        }

        /// <summary>
        /// Destructor
        /// </summary>
        ~PollingImpl()
        {
            this.Dispose();
        }

        /// <summary>
        /// Initializes the consumer builder
        /// </summary>
        private void InitializeConsumer()
        {
            //// Note: If a key or value deserializer is not set, the 
            //// deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
            //// will be used automatically (where available). The default deserializer for string
            //// is UTF8. The default deserializer for Ignore returns null for all input data (including non-null data).
            this.consumerBuilder = new ConsumerBuilder<string, byte[]>(this.consumerConfig);
            this.consumerBuilder.SetKeyDeserializer(Deserializers.Utf8);
            this.consumerBuilder.SetValueDeserializer(Deserializers.ByteArray);
            this.consumerBuilder.SetErrorHandler(this.SubscribeErrorHandler);
            //this.consumerBuilder.SetLogHandler();

            //// Build consumer
            lock (this.mutex)
            {
                this.consumer = this.consumerBuilder.Build();
            }
        }

        #endregion Constructors & Initializers


        #region Public Methods

        /// <summary>
        /// Start poll loop to consume message
        /// </summary>
        public void StartPolling()
        {
            try
            {
                Logger.Info($"{Kafka_Consumer.KAFKA_INFO_POLLING_STARTED}");
#if DEBUG
                Debug.WriteLine($"[{DateTime.Now}] {Kafka_Consumer.KAFKA_INFO_POLLING_STARTED}");
#endif

                this.StopSubscribeTimer();

                this.cancellationToken = new CancellationTokenSource();
                this.isCancelled = false;

                if (this.consumer != null)
                {
                    this.consumer.Subscribe(this.TopicName.Name);
                    this.IsConnected = true;

                    //// Consumer poll loop (endless)
                    while (this.IsConnected)
                    {
                        this.Consume();
                    }
                }
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, $"{Kafka_Consumer.KAFKA_MSG_SUBSCRIPTION_CANCELLED}");
                this.isCancelled = true;
                this.IsConnected = false;
                this.isRestartPolling = (this.cancellationToken == null) ?
                    false :
                    !this.cancellationToken.IsCancellationRequested;
            }
            catch (KafkaRetriableException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.IsConnected = false;
                this.isRestartPolling = true;
            }
            catch (ConsumeException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.IsConnected = false;
                this.isRestartPolling = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                this.IsConnected = false;
                this.isRestartPolling = false;
                this.StopSubscribeTimer();
                throw;
            }
            finally
            {
                if (this.isRestartPolling)
                {
                    this.RestartSubscribeTimer();
                }
                else
                {
                    this.StopSubscribeTimer();
                }
            }
        }

        /// <summary>
        /// Stop polling message
        /// </summary>
        public void StopPolling()
        {
            try
            {
                if (this.cancellationToken != null && !this.cancellationToken.IsCancellationRequested)
                {
                    this.cancellationToken.Cancel();
                    this.cancellationToken.Dispose();
                }

                this.cancellationToken = null;
                this.isCancelled = true;

                Logger.Info($"{Kafka_Consumer.KAFKA_INFO_POLLING_STOPPED}");
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Commit message offset
        /// </summary>
        /// <param name="brokerMessage">Kafka cluster message, contains partition and offset information for commit</param>
        public bool Commit(object brokerMessage)
        {
            try
            {
                if (brokerMessage == null)
                {
                    throw new ArgumentNullException($"{Kafka_Consumer.KAFKA_WARN_NULL_BROKER_MESSAGE}: {nameof(brokerMessage)}");
                }
                else if (!(brokerMessage is ConsumeResult<string, byte[]>))
                {
                    throw new ArgumentException($"{Kafka_Consumer.KAFKA_ERROR_INVALID_MESSAGE_TYPE}: {brokerMessage.GetType().FullName}");
                }
                else if (this.consumer == null)
                {
                    throw new ArgumentNullException($"{Kafka_Consumer.KAFKA_ERROR_SUBSCRIBER_DISPOSED}");
                }

                ConsumeResult<string, byte[]> consumeResult = brokerMessage as ConsumeResult<string, byte[]>;
                this.consumer.Commit(consumeResult);

                return true;
            }
            catch (KafkaException ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                this.SubscribeErrorHandler(this.consumer, ex.Error);
                return false;
            }
        }

        /// <summary>
        /// Dispose polling task
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Public Methods


        #region Private Methods

        /// <summary>
        /// Consume/poll for messages
        /// </summary>
        private void Consume()
        {
            this.cancellationToken.Token.ThrowIfCancellationRequested();
            ConsumeResult<string, byte[]> consumeResult = this.consumer.Consume(this.cancellationToken.Token);

            if (consumeResult.IsPartitionEOF)
            {
                //// Reached end of topic X in partition Y at offset Z
                Logger.Info($"{Kafka_Consumer.KAFKA_MSG_REACHED_PARTITION_EOF}: {consumeResult.Topic}, {Kafka_Consumer.KAFKA_INFO_LABEL_PARTITION}: {consumeResult.Partition}, {Kafka_Consumer.KAFKA_INFO_LABEL_OFFSET}: {consumeResult.Offset}.");
                return;
            }

            //// Handle message notification
            TMessage messageObject = MessageObjectSerializer.Deserialize<TMessage>(consumeResult.Message.Value);
            bool isConsumed = this.subscribeMessageHandler(this.TopicName, messageObject, consumeResult);

            //// TODO: Handle consume message success/failure state with auto commit state (default: auto commit is enabled)
            if (isConsumed)
            {
                Logger.Info($"{Kafka_Consumer.KAFKA_INFO_COMMIT_SUCCESS}");
            }
        }

        /// <summary>
        /// Set the handler to call on error events e.g. connection failures or all brokers down.
        /// Note that the client will try to automatically recover from errors that are not marked as fatal.
        /// Non-fatal errors should be interpreted as informational rather than catastrophic.
        /// </summary>
        /// <param name="iconsumer">Consumer instance</param>
        /// <param name="error">Kafka error</param>
        private void SubscribeErrorHandler(IConsumer<string, byte[]> iconsumer, Error error)
        {
            if (error == null)
            {
                throw new ArgumentNullException($"{Kafka_Consumer.KAFKA_WARN_NULL_KAFKA_ERROR}: {nameof(error)}");
            }

            if (this.IsConnected && !this.isCancelled && !this.IsDisposed)
            {
                this.RestartSubscribeTimer();
            }

            this.subscribeErrorHandler?.Invoke(error.Reason, error.IsFatal);
        }

        /// <summary>
        /// Dispose consumer
        /// </summary>
        private void DisposeConsumer()
        {
            try
            {
                lock (this.mutex)
                {
                    if (this.consumer != null)
                    {
                        this.consumer.Unsubscribe();
                        this.consumer.Close();
                        this.consumer.Dispose();

                        this.consumer = null;
                        this.consumerBuilder = null;
                    }
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing"></param>
        protected virtual void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed)
                {
                    this.DisposeConsumer();
                }

                this.IsDisposed = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }


        #region Resubscribe Timer

        /// <summary>
        /// Start re-subscribe timer
        /// </summary>
        /// <returns>Return TRUE if started else FALSE</returns>
        private bool RestartSubscribeTimer()
        {
            try
            {
                this.resubscribeTimer.Stop();
                this.resubscribeTimer.Start();

                return this.resubscribeTimer.Enabled;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
        }

        /// <summary>
        /// Stop re-subscribe timer
        /// </summary>
        /// <returns>Return TRUE if stopped else FALSE</returns>
        private bool StopSubscribeTimer()
        {
            try
            {
                this.resubscribeTimer.Stop();

                return !this.resubscribeTimer.Enabled;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                return false;
            }
        }

        /// <summary>
        /// Re-subscribe timer elapsed event
        /// </summary>
        /// <param name="sender">Trigger source</param>
        /// <param name="e">Event argument</param>
        private void ResubscribeTimerOnElapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            try
            {
                //// Log timer elapsed event
                Logger.Info($"{Kafka_Consumer.KAFKA_INFO_POLLING_RESTARTING}");
#if DEBUG
                Debug.WriteLine($"[{DateTime.Now}] {Kafka_Consumer.KAFKA_INFO_POLLING_RESTARTING}");
#endif

                this.StopPolling();
                this.StartPolling();
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        #endregion Resubscribe Timer

        #endregion Private Methods
    }
}
